/**
* Copyright (C) 2008 Happy Fish / YuQing
*
* FastDFS may be copied only under the terms of the GNU General
* Public License V3, which may be found in the FastDFS source kit.
* Please visit the FastDFS Home Page http://www.csource.org/ for more detail.
**/

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>
#include <string.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <errno.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <pthread.h>
#include <event.h>
#include "shared_func.h"
#include "logger.h"
#include "hash.h"
#include "global.h"
#include "sockopt.h"
#include "fnio_proto.h"
#include "fdht_proto_types.h"
#include "server_io.h"

/* server array, malloced when startup */
static struct fnio_server_info **pp_uniq_server_list = NULL;

/* server array organized by group (as FastDHT) */
static struct fnio_server_info ***ppp_group_server_list = NULL;

/* recv data from the server */
static int server_sock_do_read_data(struct fnio_server_info *pServerInfo);
static void server_sock_read_data(int sock, short event, void *arg);

/* proxy client to proxy server */
static void server_sock_write_data_p2p(int sock, short event, void *arg);

/* proxy server to true service/server */
static void server_sock_write_data_p2s(int sock, short event, void *arg);

/* check connect result/status (success or fail) */
static void server_sock_connect_result_check(int sock, short event, void *arg);

static int server_sock_connect_success(struct fnio_server_info *pServer);

static void close_server_sock(struct fnio_server_info *pServer);

/* used by proxy server to send data to true server */
static int pop_pkg_from_send_buff(struct fnio_server_info *pServer);
static int turn_on_server_sending(struct fnio_server_info *pServer);

/* notify proxy server to connect to true service / server */
static int fnio_notify_connect_true_server(struct fnio_server_info *pServer, \
		const int key_hash_code);

static int fnio_init_group_server_list();

int server_io_init()
{
	int result;
	struct fnio_server_info *server_list;
	struct fnio_server_info *pServerInfo;
	struct fnio_server_info *pServerEnd;
	struct fnio_server_info **ppServer;
	struct fnio_server_info **ppServerEnd;
	FDHTServerInfo *pDHTServer;

	pp_uniq_server_list = (struct fnio_server_info **)malloc( \
			g_group_array.server_count * \
			sizeof(struct fnio_server_info *));
	if (pp_uniq_server_list == NULL)
	{
		logError("file: "__FILE__", line: %d, " \
			"malloc %d bytes fail, " \
			"errno: %d, error info: %s", \
			__LINE__, g_group_array.server_count * \
                        sizeof(struct fnio_server_info *), \
			errno, strerror(errno));
		return errno != 0 ? errno : ENOMEM;
	}

	result = 0;
	memset(pp_uniq_server_list, 0, g_group_array.server_count * \
		sizeof(struct fnio_server_info *));

	ppServerEnd = pp_uniq_server_list + g_group_array.server_count;
	for (ppServer=pp_uniq_server_list, pDHTServer=g_group_array.servers; \
		ppServer<ppServerEnd; ppServer++, pDHTServer++)
	{
		server_list = (struct fnio_server_info *)malloc( \
			g_server_work_threads * sizeof(struct fnio_server_info));
		if (server_list == NULL)
		{
			logError("file: "__FILE__", line: %d, " \
				"malloc %d bytes fail, " \
				"errno: %d, error info: %s", \
				__LINE__, g_server_work_threads * \
				sizeof(struct fnio_server_info), \
				errno, strerror(errno));
			return errno != 0 ? errno : ENOMEM;
		}

		memset(server_list, 0, g_server_work_threads * \
			sizeof(struct fnio_server_info));

		pServerEnd = server_list + g_server_work_threads;
		for (pServerInfo=server_list; pServerInfo<pServerEnd; \
				pServerInfo++)
		{
			strcpy(pServerInfo->ip_addr, pDHTServer->ip_addr);
			pServerInfo->port = pDHTServer->port;
			pServerInfo->sock = -1;
			result = fnio_malloc_send_buff(&(pServerInfo->sbuff));
			if (result != 0)
			{
				return result;
			}

			pServerInfo->pRecvNode = free_queue_pop();
			if (pServerInfo->pRecvNode == NULL)
			{
				logError("file: "__FILE__", line: %d, " \
					"malloc task buff failed", \
					__LINE__);
				return ENOSPC;
			}

			pServerInfo->pSendNode = free_queue_pop();
			if (pServerInfo->pSendNode == NULL)
			{
				logError("file: "__FILE__", line: %d, " \
					"malloc task buff failed", \
					__LINE__);

				return ENOSPC;
			}
		}

		*ppServer = server_list;
	}

	return fnio_init_group_server_list();

	/*
	for (ppServer=pp_uniq_server_list; ppServer<ppServerEnd; ppServer++)
	{
		pServerEnd = *ppServer + g_server_work_threads;
		for (pServerInfo=*ppServer; pServerInfo<pServerEnd; \
				pServerInfo++)
		{
			fnio_check_server_connected(pServerInfo);
		}
	}

	return result;
	*/
}

static int fnio_init_group_server_list()
{
	int group_id;
	ServerArray *pServerArray;
	FDHTServerInfo **ppServer;
	FDHTServerInfo **ppServerEnd;
	int server_index;
	int i;

	ppp_group_server_list = (struct fnio_server_info ***)malloc( \
			g_group_array.group_count * \
			sizeof(struct fnio_server_info **));
	if (ppp_group_server_list == NULL)
	{
		logError("file: "__FILE__", line: %d, " \
			"malloc %d bytes fail, " \
			"errno: %d, error info: %s", \
			__LINE__, g_group_array.group_count * \
                        sizeof(struct fnio_server_info **), \
			errno, strerror(errno));
		return errno != 0 ? errno : ENOMEM;
	}

	pServerArray = g_group_array.groups;
	for (group_id=0; group_id<g_group_array.group_count; group_id++)
	{
		ppp_group_server_list[group_id] = (struct fnio_server_info **)
				malloc(pServerArray->count * \
				sizeof(struct fnio_server_info *));
		if (ppp_group_server_list[group_id] == NULL)
		{
			logError("file: "__FILE__", line: %d, " \
					"malloc %d bytes fail, " \
					"errno: %d, error info: %s", \
					__LINE__, pServerArray->count * \
					sizeof(struct fnio_server_info *), \
					errno, strerror(errno));
			return errno != 0 ? errno : ENOMEM;
		}

		ppServerEnd = pServerArray->servers + pServerArray->count;
		for (ppServer=pServerArray->servers, i=0; \
				ppServer<ppServerEnd; ppServer++, i++)
		{
			server_index = *ppServer - g_group_array.servers;
			ppp_group_server_list[group_id][i] = \
					pp_uniq_server_list[server_index];
		}

		pServerArray++;
	}

	return 0;
}

static void fnio_free_server(struct fnio_server_info *server_list)
{
	struct fnio_server_info *pServerInfo;
	struct fnio_server_info *pServerEnd;

	if (server_list == NULL)
	{
		return;
	}

	pServerEnd = server_list + g_server_work_threads;
	for (pServerInfo=server_list; pServerInfo<pServerEnd; pServerInfo++)
	{
		if (pServerInfo->sbuff.send_buff != NULL)
		{
			free(pServerInfo->sbuff.send_buff);
		}

		if (pServerInfo->pSendNode != NULL)
		{
			free_queue_push(pServerInfo->pSendNode);
		}

		if (pServerInfo->pRecvNode != NULL)
		{
			free_queue_push(pServerInfo->pRecvNode);
		}
	}

	free(server_list);
}

#define FNIO_GET_SERVER_BY_HASH_CODE(key_hash_code, new_hash_code, \
		pServerArray, pServer, group_id, server_index, select_index) \
	group_id = ((unsigned int)key_hash_code) % g_group_array.group_count; \
	pServerArray = g_group_array.groups + group_id; \
\
	new_hash_code = (key_hash_code << 16) | (key_hash_code >> 16); \
	server_index = new_hash_code % pServerArray->count; \
\
	select_index = (unsigned int)key_hash_code % g_server_work_threads; \
	pServer = ppp_group_server_list[group_id][server_index] + select_index;


int fnio_check_server_connected_by_hash_code(const int key_hash_code)
{
	ServerArray *pServerArray;
	struct fnio_server_info *pServer;
	int group_id;
	int server_index;
	int select_index;
	unsigned int new_hash_code;

	FNIO_GET_SERVER_BY_HASH_CODE(key_hash_code, new_hash_code, \
		pServerArray, pServer, group_id, server_index, select_index)

	return fnio_check_server_connected(pServer);
}

struct fnio_server_info *fnio_get_server(const int key_hash_code)
{
	ServerArray *pServerArray;
	struct fnio_server_info *pServer;
	int group_id;
	int server_index;
	int select_index;
	unsigned int new_hash_code;
	int i;
	int result;

	FNIO_GET_SERVER_BY_HASH_CODE(key_hash_code, new_hash_code, \
		pServerArray, pServer, group_id, server_index, select_index)

	/*
	logInfo("key_hash_code=%d, group_id=%d, server_index=%d, last_conn_errno=%d,  connected=%d", \
			key_hash_code, group_id, server_index,  pServer->last_conn_errno, pServer->connected);
	*/

	result = fnio_check_server_connected(pServer);
	if ((result == 0 && pServer->p2s_last_conn_errno == 0) || \
		(result == EINPROGRESS && pServer->last_conn_errno == 0))
	{
		return pServer;
	}
	else if (g_server_mode == FNIO_SERVER_MODE_CLIENT && \
		pServer->p2s_last_conn_errno != 0)
	{
		fnio_notify_connect_true_server(pServer, key_hash_code);
	}

	if (pServerArray->count == 1)
	{
		return NULL;
	}

	for (i=0; i<server_index; i++)
	{
		pServer = ppp_group_server_list[group_id][i] + select_index;
		result = fnio_check_server_connected(pServer);
		if ((result == 0 && pServer->p2s_last_conn_errno == 0) || \
			(result==EINPROGRESS && pServer->last_conn_errno == 0))
		{
			return pServer;
		}
		else if (g_server_mode == FNIO_SERVER_MODE_CLIENT && \
			pServer->p2s_last_conn_errno != 0)
		{
			fnio_notify_connect_true_server(pServer, key_hash_code);
		}
	}

	for (i=server_index+1; i<pServerArray->count; i++)
	{
		pServer = ppp_group_server_list[group_id][i] + select_index;
		result = fnio_check_server_connected(pServer);
		if ((result == 0 && pServer->p2s_last_conn_errno == 0) || \
			(result==EINPROGRESS && pServer->last_conn_errno == 0))
		{
			return pServer;
		}
		else if (g_server_mode == FNIO_SERVER_MODE_CLIENT && \
			pServer->p2s_last_conn_errno != 0)
		{
			fnio_notify_connect_true_server(pServer, key_hash_code);
		}
	}

	return NULL;
}

int fnio_push_to_server_send_buff(struct fnio_server_info *pServer, \
			struct fnio_client_info *pClientInfo)
{
	FNIOProtoHeader header;
	int result;

	int2buff(pClientInfo->pRecvNode->length, header.pkg_len);
	long2buff(pClientInfo->session_id, header.session_id);
	header.source = FNIO_PROTO_SOURCE_DATA;

	result = fnio_push_to_send_buff(&pServer->sbuff,
		(char *)&header, sizeof(FNIOProtoHeader), 
		pClientInfo->pRecvNode->data, pClientInfo->pRecvNode->length);

	if (result != 0)
	{
		logError("file: "__FILE__", line: %d, " \
			"call fnio_push_to_send_buff fail, " \
			"send buff size: %d, used bytes: %d, "  \
			"errno: %d, error info: %s", \
			__LINE__, pServer->sbuff.buff_end - \
			pServer->sbuff.send_buff, \
			pServer->sbuff.send_bytes, \
			result, strerror(result));
		return result;
	}

	result = turn_on_server_sending(pServer);
	return result == EINPROGRESS ? 0 : result;
}

static void close_server_sock(struct fnio_server_info *pServer)
{
	event_del(&pServer->ev_read);
	if (pServer->sbuff.send_bytes > 0)
	{
		event_del(&pServer->ev_write);
	}

	if (pServer->sock >= 0)
	{
		close(pServer->sock);
		pServer->sock = -1;
		if (pServer->last_conn_errno == 0)
		{
			pServer->last_conn_errno = ENOTCONN;
		}

		if (g_server_mode == FNIO_SERVER_MODE_SERVER)
		{
			fnio_notify_server_conn_status(pServer);
		}
	}

	pServer->connected = false;
	pServer->sending = false;
}

int fnio_check_server_connected(struct fnio_server_info *pServer)
{
	int port;
	int result;

	if (pServer->connected)
	{
		return 0;
	}

	if (pServer->sock >= 0)
	{
		return EINPROGRESS;
	}

	pServer->sock = socket(AF_INET, SOCK_STREAM, 0);
	if(pServer->sock < 0)
	{
		pServer->last_conn_errno = errno != 0 ? errno : EMFILE;
		logError("file: "__FILE__", line: %d, " \
			"socket create fail, " \
			"errno: %d, error info: %s", __LINE__, \
			pServer->last_conn_errno, \
			strerror(pServer->last_conn_errno));
		return pServer->last_conn_errno;
	}

	if ((result=tcpsetnonblockopt(pServer->sock, \
			g_network_timeout)) != 0)
	{
		pServer->last_conn_errno = result;
		close_server_sock(pServer);
		return pServer->last_conn_errno;
	}

	if (g_server_mode == FNIO_SERVER_MODE_CLIENT)
	{
		port = g_proxy_server_port;
	}
	else
	{
		port = pServer->port;
	}

	if ((result=connectserverbyip(pServer->sock, \
		pServer->ip_addr, port)) != 0)
	{
		if (result != EINPROGRESS)
		{
			pServer->last_conn_errno = result;
			close_server_sock(pServer);
			return pServer->last_conn_errno;
		}
	}

	event_set(&pServer->ev_write, pServer->sock, EV_WRITE, \
		server_sock_connect_result_check, pServer);

	result = event_base_set(g_event_base, \
			&pServer->ev_write);
	if (result != 0)
	{
		pServer->last_conn_errno = result;
		close_server_sock(pServer);

		logError("file: "__FILE__", line: %d, " \
			"event_base_set fail, errno: %d, error info: %s", \
			__LINE__, pServer->last_conn_errno, \
			strerror(pServer->last_conn_errno));
		return pServer->last_conn_errno;
	}
	if ((result=event_add(&pServer->ev_write, \
			&g_network_tv)) != 0)
	{
		pServer->last_conn_errno = result;
		close_server_sock(pServer);

		logError("file: "__FILE__", line: %d, " \
			"event_add fail, errno: %d, error info: %s", \
			__LINE__, pServer->last_conn_errno, \
			strerror(pServer->last_conn_errno));
		return pServer->last_conn_errno;
	}

	return EINPROGRESS;
}

static int reconnect_server(struct fnio_server_info *pServer)
{
	close_server_sock(pServer);
	return fnio_check_server_connected(pServer);
}

static void server_sock_connect_result_check(int sock, short event, void *arg)
{
	struct fnio_server_info *pServer;
	struct sockaddr_in addr;
	socklen_t addrlen;

	pServer = (struct fnio_server_info *)arg;
	if (event == EV_TIMEOUT)
	{
		pServer->last_conn_errno = ETIMEDOUT;
		logError("file: "__FILE__", line: %d, " \
			"connect to %s:%d timeout", __LINE__, \
			pServer->ip_addr, pServer->port);

		close_server_sock(pServer);
		return;
	}

	addrlen = sizeof(addr);
	if (getpeername(sock, (struct sockaddr *)&addr, &addrlen) == 0)
	{
		server_sock_connect_success(pServer);
		return;
	}

	recv(pServer->sock, pServer->pRecvNode->data, 1, 0);
	pServer->last_conn_errno = errno != 0 ? errno : EIO;
	logError("file: "__FILE__", line: %d, " \
		"connect to %s:%d fail, errno: %d, error info: %s", \
		__LINE__, pServer->ip_addr, \
		g_server_mode == FNIO_SERVER_MODE_CLIENT ? \
		g_proxy_server_port : pServer->port, \
		pServer->last_conn_errno, strerror(pServer->last_conn_errno));

	close_server_sock(pServer);
	return;
}

static int server_sock_connect_success(struct fnio_server_info *pServer)
{
	int result;

	if (g_server_mode == FNIO_SERVER_MODE_CLIENT)
	{
		event_set(&pServer->ev_write, pServer->sock, EV_WRITE, \
			server_sock_write_data_p2p, pServer);
	}
	else
	{
		event_set(&pServer->ev_write, pServer->sock, EV_WRITE, \
			server_sock_write_data_p2s, pServer);
	}

	pServer->pSendNode->length = 0;
	pServer->pSendNode->offset = 0;

	pServer->pRecvNode->length = 0;
	pServer->pRecvNode->offset = 0;
	result = event_base_set(g_event_base, \
				&pServer->ev_write);
	if (result != 0)
	{
		pServer->last_conn_errno = result;
		close_server_sock(pServer);

		logError("file: "__FILE__", line: %d, " \
			"event_base_set fail, errno: %d, error info: %s", \
			__LINE__, pServer->last_conn_errno, \
			strerror(pServer->last_conn_errno));
		return pServer->last_conn_errno;
	}
	if (pServer->sbuff.send_bytes > 0)
	{
		if ((result=event_add(&pServer->ev_write, \
				&g_network_tv)) != 0)
		{
			pServer->last_conn_errno = result;
			close_server_sock(pServer);

			logError("file: "__FILE__", line: %d, " \
				"event_add fail, errno: %d, error info: %s", \
				__LINE__, pServer->last_conn_errno, \
				strerror(pServer->last_conn_errno));
			return pServer->last_conn_errno;
		}

		pServer->sending = true;
	}

	event_set(&pServer->ev_read, pServer->sock, EV_READ, \
		server_sock_read_data, pServer);
	result = event_base_set(g_event_base, \
			&pServer->ev_read);
	if (result != 0)
	{
		pServer->last_conn_errno = result;
		close_server_sock(pServer);

		logError("file: "__FILE__", line: %d, " \
			"event_base_set fail, errno: %d, error info: %s", \
			__LINE__, pServer->last_conn_errno, \
			strerror(pServer->last_conn_errno));
		return pServer->last_conn_errno;
	}

	pServer->last_conn_errno = 0;
	if (g_server_mode == FNIO_SERVER_MODE_CLIENT)
	{
		if ((result=event_add(&pServer->ev_read, \
				&g_network_tv)) != 0)
		{
			pServer->last_conn_errno = result;
			close_server_sock(pServer);

			logError("file: "__FILE__", line: %d, " \
				"event_add fail, errno: %d, error info: %s", \
				__LINE__, pServer->last_conn_errno, \
				strerror(pServer->last_conn_errno));
			return pServer->last_conn_errno;
		}
	}
	else
	{
		fnio_notify_server_conn_status(pServer);
	}

	pServer->connected = true;
	logInfo("file: "__FILE__", line: %d, " \
		"success connect to %s:%d", __LINE__, \
		pServer->ip_addr, g_server_mode == FNIO_SERVER_MODE_CLIENT ? \
		g_proxy_server_port : pServer->port);

	return 0;
}

static void server_sock_write_data_p2p(int sock, short event, void *arg)
{
	struct fnio_server_info *pServer;
	int result;
	int bytes;
	int send_bytes;

	pServer = (struct fnio_server_info *)arg;
	if (event == EV_TIMEOUT)
	{
		logError("file: "__FILE__", line: %d, " \
			"write to %s:%d timeout", __LINE__, \
			pServer->ip_addr, g_proxy_server_port);

		if ((result=event_add(&pServer->ev_write, \
					&g_network_tv)) != 0)
		{
			pServer->sending = false;

			logError("file: "__FILE__", line: %d, " \
				"event_add fail, " \
				"errno: %d, error info: %s", \
				__LINE__, result, strerror(result));
		}

		return;
	}

	do
	{
	if (pServer->sbuff.send_bytes == 0)
	{
		pServer->sending = false;
		break;
	}

	if (pServer->sbuff.send_start + pServer->sbuff.send_bytes <=  pServer->sbuff.buff_end)
	{
		send_bytes = pServer->sbuff.send_bytes;
	}
	else
	{
		send_bytes = pServer->sbuff.buff_end - pServer->sbuff.send_start;
	}

	bytes = send(sock, pServer->sbuff.send_start, send_bytes, 0);
	if (bytes < 0)
	{
		if (errno == EAGAIN || errno == EWOULDBLOCK)
		{
			if ((result=event_add(&pServer->ev_write, \
					&g_network_tv)) != 0)
			{
				pServer->sending = false;

				logError("file: "__FILE__", line: %d, " \
					"event_add fail, " \
					"errno: %d, error info: %s", \
					__LINE__, result, strerror(result));
			}
		}
		else
		{
			reconnect_server(pServer);
			logError("file: "__FILE__", line: %d, " \
				"write to %s:%d fail, " \
				"errno: %d, error info: %s", \
				__LINE__, pServer->ip_addr, \
				g_proxy_server_port, \
				errno, strerror(errno));
		}

		break;
	}
	else if (bytes == 0)
	{
		reconnect_server(pServer);

		logError("file: "__FILE__", line: %d, " \
			"send to %s:%d failed, connection disconnected.", \
			__LINE__, pServer->ip_addr, g_proxy_server_port);
		break;
	}

	pServer->sbuff.send_bytes -= bytes;
	pServer->sbuff.send_start += bytes;
	if (pServer->sbuff.send_start >= pServer->sbuff.buff_end)
	{
		pServer->sbuff.send_start = pServer->sbuff.send_buff;
	}

	if (pServer->sbuff.send_bytes > 0)
	{
		if ((result=event_add(&pServer->ev_write, &g_network_tv)) != 0)
		{
			pServer->sending = false;
			logError("file: "__FILE__", line: %d, " \
				"event_add fail, " \
				"errno: %d, error info: %s", \
				__LINE__, result, strerror(result));
		}
	}
	else
	{
		pServer->sending = false;
	}
	} while (0);
}

static int get_pkg_header_from_send_buff(struct fnio_server_info *pServer, \
		char **ppStart, int *pkg_len, char *session_id)
{
	int byte1;
	int byte2;
	FNIOProtoHeader *pHeader;

	if (*ppStart + sizeof(FNIOProtoHeader) <= pServer->sbuff.buff_end)
	{
		pHeader = (FNIOProtoHeader *)*ppStart;
		*ppStart += sizeof(FNIOProtoHeader);
	}
	else
	{
		byte1 = pServer->sbuff.buff_end - *ppStart;
		byte2 = sizeof(FNIOProtoHeader) - byte1;
		memcpy(pServer->pSendNode->data, *ppStart, byte1);
		memcpy(pServer->pSendNode->data+byte1, 
			pServer->sbuff.send_buff, byte2);
		pHeader = (FNIOProtoHeader *)pServer->pSendNode->data;
		*ppStart = pServer->sbuff.send_buff + byte2;
	}

	*pkg_len = buff2int(pHeader->pkg_len);
	if (pServer->sbuff.send_bytes < sizeof(FNIOProtoHeader) + (*pkg_len))
	{
		return EINVAL;
	}

	memcpy(session_id, pHeader->session_id, 8);
	return 0;
}

static int pop_pkg_from_send_buff(struct fnio_server_info *pServer)
{
	int byte1;
	int byte2;
	int pkg_len;
	int result;
	char *pStart;

	if (pServer->sbuff.send_bytes < 2 * sizeof(FNIOProtoHeader))
	{
		pServer->sending = false;
		return ENOENT;
	}

	pStart = pServer->sbuff.send_start;
	if ((result=get_pkg_header_from_send_buff(pServer, &pStart, &pkg_len, \
			pServer->server_session_id)) != 0)
	{
		pServer->sending = false;
		return result;
	}

	if ((result=get_pkg_header_from_send_buff(pServer, &pStart, &pkg_len, \
			pServer->client_session_id)) != 0)
	{
		pServer->sending = false;
		return result;
	}

	if (pkg_len > pServer->pSendNode->size)
	{
		char *pNewBuff;

		pNewBuff = (char *)malloc(pkg_len);
		if (pNewBuff == NULL)
		{
			logError("file: "__FILE__", line: %d, " \
				"malloc %d bytes failed, " \
				"errno: %d, error info: %s", \
				__LINE__, pkg_len, errno, strerror(errno));

			pServer->sending = false;
			return errno != 0 ? errno : ENOMEM;
		}

		free(pServer->pSendNode->data);
		pServer->pSendNode->data = pNewBuff;
		pServer->pSendNode->size = pkg_len;
	}

	pServer->sbuff.send_start = pStart;
	if (pServer->sbuff.send_start + pkg_len <= pServer->sbuff.buff_end)
	{
		memcpy(pServer->pSendNode->data, pServer->sbuff.send_start, pkg_len);
		pServer->sbuff.send_start += pkg_len;
	}
	else
	{
		byte1 = pServer->sbuff.buff_end - pServer->sbuff.send_start;
		byte2 = pkg_len - byte1;
		memcpy(pServer->pSendNode->data, pServer->sbuff.send_start, byte1);
		memcpy(pServer->pSendNode->data+byte1, 
			pServer->sbuff.send_buff, byte2);
		pServer->sbuff.send_start = pServer->sbuff.send_buff + byte2;
	}

	pServer->pSendNode->length = pkg_len;
	pServer->pSendNode->offset = 0;
	pServer->sbuff.send_bytes -= 2 * sizeof(FNIOProtoHeader) + pkg_len;

	return 0;
}

static void server_sock_write_data_p2s(int sock, short event, void *arg)
{
	struct fnio_server_info *pServer;
	int result;
	int bytes;

	pServer = (struct fnio_server_info *)arg;
	if (event == EV_TIMEOUT)
	{
		pServer->sending = false;
		logError("file: "__FILE__", line: %d, " \
			"write to %s:%d timeout", __LINE__, \
			pServer->ip_addr, pServer->port);

		if ((result=event_add(&pServer->ev_write, \
					&g_network_tv)) != 0)
		{
			pServer->sending = false;
			logError("file: "__FILE__", line: %d, " \
				"event_add fail, " \
				"errno: %d, error info: %s", \
				__LINE__, result, strerror(result));
		}

		return;
	}
	
	if (pServer->pSendNode->length == 0)
	{
		if (pop_pkg_from_send_buff(pServer) != 0)
		{
			return;
		}
	}

	bytes = send(sock, pServer->pSendNode->data + pServer->pSendNode->offset, 
			pServer->pSendNode->length - pServer->pSendNode->offset, 0);
	if (bytes < 0)
	{
		if (errno == EAGAIN || errno == EWOULDBLOCK)
		{
			if ((result=event_add(&pServer->ev_write, \
					&g_network_tv)) != 0)
			{
				pServer->sending = false;
				logError("file: "__FILE__", line: %d, " \
					"event_add fail, " \
					"errno: %d, error info: %s", \
					__LINE__, result, strerror(result));
			}
		}
		else
		{
			reconnect_server(pServer);
			logError("file: "__FILE__", line: %d, " \
				"write to %s:%d fail, " \
				"errno: %d, error info: %s", \
				__LINE__, pServer->ip_addr, pServer->port, \
				errno, strerror(errno));
		}

		return;
	}
	else if (bytes == 0)
	{
		reconnect_server(pServer);

		logWarning("file: "__FILE__", line: %d, " \
			"send to %s:%d failed, connection disconnected", \
			__LINE__, pServer->ip_addr, pServer->port);
		return;
	}

	//printf("send to server %d bytes, total bytes: %d\n", bytes, pServer->pSendNode->length);

	pServer->pSendNode->offset += bytes;
	if (pServer->pSendNode->offset < pServer->pSendNode->length)
	{
		if ((result=event_add(&pServer->ev_write, &g_network_tv)) != 0)
		{
			pServer->sending = false;
			logError("file: "__FILE__", line: %d, " \
				"event_add fail, " \
				"errno: %d, error info: %s", \
				__LINE__, result, strerror(result));
		}

		return;
	}

	pServer->pSendNode->length = 0;
	pServer->pSendNode->offset = 0;
	
	if ((result=event_add(&pServer->ev_read, \
			&g_network_tv)) != 0)
	{
		close_server_sock(pServer);

		logError("file: "__FILE__", line: %d, " \
			"event_add fail, errno: %d, error info: %s", \
			__LINE__, result, strerror(result));
		return;
	}
}

static int server_sock_do_read_data(struct fnio_server_info *pServerInfo)
{
	int bytes;
	int recv_bytes;
	int header_size;

	if (g_server_mode == FNIO_SERVER_MODE_CLIENT)
	{
		header_size = sizeof(FNIOProtoHeader);
	}
	else
	{
		header_size = sizeof(FDHTProtoHeader);
	}

	if (pServerInfo->pRecvNode->length == 0) //recv header
	{
		recv_bytes = header_size - pServerInfo->pRecvNode->offset;
	}
	else
	{
		recv_bytes = pServerInfo->pRecvNode->length - \
				pServerInfo->pRecvNode->offset;
	}

	if (pServerInfo->pRecvNode->offset + recv_bytes > \
			pServerInfo->pRecvNode->size)
	{
		char *pTemp;

		pTemp = pServerInfo->pRecvNode->data;
		pServerInfo->pRecvNode->data = (char *)realloc( \
				pServerInfo->pRecvNode->data, \
				pServerInfo->pRecvNode->size + recv_bytes);
		if (pServerInfo->pRecvNode->data == NULL)
		{
			logError("file: "__FILE__", line: %d, " \
				"malloc failed, " \
				"errno: %d, error info: %s", \
				__LINE__, errno, strerror(errno));

			pServerInfo->pRecvNode->data = pTemp;  //restore old data
			return errno != 0 ? errno : ENOMEM;
		}

		pServerInfo->pRecvNode->size += recv_bytes;
	}

	bytes = recv(pServerInfo->sock, pServerInfo->pRecvNode->data + \
			pServerInfo->pRecvNode->offset, recv_bytes, 0);
	if (bytes < 0)
	{
		if (!(errno == EAGAIN || errno == EWOULDBLOCK))
		{
			logError("file: "__FILE__", line: %d, " \
				"recv from %s:%d failed, " \
				"errno: %d, error info: %s", \
				__LINE__, pServerInfo->ip_addr, \
				g_server_mode == FNIO_SERVER_MODE_CLIENT ? \
				g_proxy_server_port : pServerInfo->port, \
				errno, strerror(errno));
		}

		return errno;
	}
	else if (bytes == 0)
	{
		logWarning("file: "__FILE__", line: %d, " \
			"connection closed by peer %s:%d", \
			__LINE__, pServerInfo->ip_addr, 
			g_server_mode == FNIO_SERVER_MODE_CLIENT ? \
			g_proxy_server_port : pServerInfo->port);
		
		return ECONNRESET;
	}

	if (pServerInfo->pRecvNode->length == 0) //header
	{
		if (pServerInfo->pRecvNode->offset+bytes < header_size)
		{
			pServerInfo->pRecvNode->offset += bytes;
			return EAGAIN;
		}

		if (g_server_mode == FNIO_SERVER_MODE_CLIENT)
		{
			pServerInfo->pRecvNode->length=buff2int(((FNIOProtoHeader*)
					pServerInfo->pRecvNode->data)->pkg_len);
		}
		else
		{
			pServerInfo->pRecvNode->length=buff2int(((FDHTProtoHeader*)
					pServerInfo->pRecvNode->data)->pkg_len);
		}

		if (pServerInfo->pRecvNode->length < 0)
		{
			logError("file: "__FILE__", line: %d, " \
				"pkg length: %d < 0", \
				__LINE__, pServerInfo->pRecvNode->length);

			return EINVAL;
		}

		pServerInfo->pRecvNode->length += header_size;
		if (pServerInfo->pRecvNode->length > g_max_buff_size)
		{
			logError("file: "__FILE__", line: %d, " \
				"pkg length: %d > max pkg size: %d", __LINE__, \
				pServerInfo->pRecvNode->length, g_max_buff_size);

			return EINVAL;
		}
	}

	pServerInfo->pRecvNode->offset += bytes;
	if (pServerInfo->pRecvNode->offset >= pServerInfo->pRecvNode->length) //recv one pkg done
	{
		return 0;
	}

	return EAGAIN;
}

static void server_sock_read_data(int sock, short event, void *arg)
{
	struct fnio_server_info *pServer;
	int result;

	pServer= (struct fnio_server_info *)arg;
	if (event == EV_TIMEOUT)
	{
		if (g_server_mode == FNIO_SERVER_MODE_CLIENT)
		{
			if (pServer->pRecvNode->offset > 0)
			{
				logError("file: "__FILE__", line: %d, " \
					"recv from %s:%d timeout", \
					__LINE__, pServer->ip_addr, \
					g_proxy_server_port);
				close_server_sock(pServer);
				return;
			}

			if ((result=event_add(&pServer->ev_read, \
					&g_network_tv)) != 0)
			{
				logError("file: "__FILE__", line: %d, " \
					"event_add fail, " \
					"errno: %d, error info: %s", \
					__LINE__, result, strerror(result));
				close_server_sock(pServer);
			}
		}
		else
		{
			logError("file: "__FILE__", line: %d, " \
				"recv from %s:%d timeout", \
				__LINE__, pServer->ip_addr, pServer->port);
			close_server_sock(pServer);
		}

		return;
	}

	result = server_sock_do_read_data(pServer);
	if (result == 0) //recv one pkg done
	{
		if (g_server_mode == FNIO_SERVER_MODE_CLIENT && \
			((FNIOProtoHeader *)pServer->pRecvNode->data)->source \
			== FNIO_PROTO_SOURCE_PROXY)
		{
			pServer->p2s_last_conn_errno = 
					*(pServer->pRecvNode->data + 
					sizeof(FNIOProtoHeader));
			//logInfo("pServer->p2s_last_conn_errno=%d", pServer->p2s_last_conn_errno);
		}
		else if (fnio_push_to_client_send_buff(pServer) != 0)
		{
		}

		pServer->pRecvNode->length = 0;
		pServer->pRecvNode->offset = 0;

		if (g_server_mode == FNIO_SERVER_MODE_SERVER)
		{
			if (pServer->sbuff.send_bytes > 0)
			{
			if ((result=event_add(&pServer->ev_write, &g_network_tv)) != 0)
			{
				logError("file: "__FILE__", line: %d, " \
						"event_add fail, " \
						"errno: %d, error info: %s", \
						__LINE__, result, strerror(result));
				pServer->sending = false;
			}
			}
			else
			{
				pServer->sending = false;
			}

			return;
		}
	}
	else if (!(result == EAGAIN || result == EWOULDBLOCK))
	{
		close_server_sock(pServer);
		return;
	}

	if ((result=event_add(&pServer->ev_read, &g_network_tv)) != 0)
	{
		logError("file: "__FILE__", line: %d, " \
			"event_add fail, errno: %d, error info: %s", \
			__LINE__, result, strerror(result));
		close_server_sock(pServer);
	}
}

static int turn_on_server_sending(struct fnio_server_info *pServer)
{
	int result;

	if (pServer->sending)
	{
		return 0;
	}

	result = fnio_check_server_connected(pServer);
	if (result != 0)
	{
		return result;
	}

	if ((result=event_add(&pServer->ev_write, \
			&g_network_tv)) != 0)
	{
		close_server_sock(pServer);
		logError("file: "__FILE__", line: %d, "\
			"event_add fail, errno: %d, " \
			"error info: %s", __LINE__, \
			result, strerror(result));
	}
	else
	{
		pServer->sending = true;
	}

	return result;
}

static int fnio_notify_connect_true_server(struct fnio_server_info *pServer, \
		const int key_hash_code)
{
	char buff[sizeof(FNIOProtoHeader)+4];
	int result;

	memset(buff, 0, sizeof(buff));
	((FNIOProtoHeader *)buff)->source = FNIO_PROTO_SOURCE_PROXY;
	int2buff(4, ((FNIOProtoHeader *)buff)->pkg_len);
	int2buff(key_hash_code, buff + sizeof(FNIOProtoHeader));

	if ((result=fnio_push_to_send_buff(&pServer->sbuff, NULL, 0, \
		buff, sizeof(buff))) == 0)
	{
		turn_on_server_sending(pServer);
	}

	return result;
}

